package com.gaogzhen.threadPool;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 阻塞队列
 * @param <T>       泛型-任务
 */
@Slf4j(topic = "c.BlockingQueue")
public class BlockingQueue<T> {
    /** 任务队列  */
    private final Deque<T> queue = new ArrayDeque<>();

    /** 锁 */
    private final ReentrantLock lock = new ReentrantLock();

    /** 条件变量-任务队列满时生产者 */
    private final Condition fullWaitSet = lock.newCondition();
    /** 条件变量-任务队列为空时消费者 */
    private final Condition emptyWaitSet = lock.newCondition();
    /** 容量 */
    private int capacity;

    public BlockingQueue(int capacity) {
        this.capacity = capacity;
    }

    /**
     * 带超时时间的阻塞获取
     * @param timeout       超时时间
     * @param unit          超时单位
     * @return              任务
     */
    public T pool(long timeout, TimeUnit unit) {
        lock.lock();
        // 超时时间统一转换为纳秒
        long nanos = unit.toNanos(timeout);
        try {
            while (queue.isEmpty()) {
                if (nanos <= 0) {
                    return null;
                }
                nanos = emptyWaitSet.awaitNanos(nanos);
            }
            T t = queue.removeFirst();
            fullWaitSet.signal();
            return t;
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
        return null;
    }

    /**
     * 阻塞获取
     * @return      任务
     */
    public T take() {
        lock.lock();
        T t = null;
        try {
            while (queue.isEmpty()) {
                emptyWaitSet.await();
            }
            t = queue.removeFirst();
            fullWaitSet.signal();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
        return t;
    }

    /**
     * 超时阻塞添加
     * @param t             任务
     * @param timeout       超时时间
     * @param unit          超时单位
     * @return              添加成功true,false添加失败
     */
    public boolean offer(T t, long timeout, TimeUnit unit) {
        lock.lock();
        // 超时时间统一转换为纳秒
        long nanos = unit.toNanos(timeout);
        try {
            while (queue.size() == capacity) {
                if (nanos <= 0) {
                    return false;
                }
                log.debug("等待加入任务队列 {}", t);
                nanos = fullWaitSet.awaitNanos(nanos);
            }
            queue.addLast(t);
            log.debug("加入任务队列 {}", t);
            emptyWaitSet.signal();
            return true;
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
        return false;
    }
    /**
     * 阻塞添加
     * @param t     任务
     */
    public void put(T t) {
        lock.lock();
        try {
            while (queue.size() == capacity) {
                log.debug("等待加入任务队列 {}", t);
                fullWaitSet.await();
            }
            queue.addLast(t);
            log.debug("加入任务队列 {}", t);
            emptyWaitSet.signal();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 获取容量
     * @return      容量
     */
    public int size() {
        lock.lock();
        try {
            return queue.size();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 尝试添加
     * @param rejectStrategy        决绝策略
     * @param t                  任务
     */
    public void tryPut(RejectStrategy<T> rejectStrategy, T t) {
        lock.lock();
        try {
            // 判断队列是否已满
            if (queue.size() == capacity) {
                rejectStrategy.reject(this, t);
            } else {
                queue.addLast(t);
                log.debug("加入任务队列 {}", t);
                emptyWaitSet.signal();
            }

        } finally {
            lock.unlock();
        }
    }
}
